Implement Kafka Module with AWS Secrets Manager in NodeJS with TypeScript [Modularized Code]

Implement Kafka Module with AWS Secrets Manager in NodeJS with TypeScript [Modularized Code]

The best way to use Kafka in your NodeJS project is to create an independent module. Here we will create an independent Kafka module that you can use in your multiple module in your appliction. In this module we will write the reusable code so that we can use it just by importing the Kafka module.

What is Apache Kafka?

Apache Kafka is mainly used for Messaging Queues. Apache Kafka is a distributed Publish-Subscribe Messaging System. Kafka was developed at LinkedIn and later became part of the Apache project. Kafka is also used for real-time data streams and to collect big data for real-time analytics.

Kafka decouples the data pipeline and solves the complexity problem. The application producing the messages to Kafka could be anything like front end server, Hadoop server, database server, or Chat server. These are called Producers.
The application which is consuming those messages from Kafka like a database server your security system, monitoring system, other services, or data warehouse is called Consumers.
So your producer sends data to Kafka, Kafka stores those messages, and the consumer who wants them can subscribe and receive them. You can have multiple subscribers to a single category of messages.
Adding a new consumer is very easy. You can go ahead and add a new consumer and subscribe to the message categories that are required.
The Kafka cluster is distributed and has multiple machines running in parallel. This is the reason why Kafka is fast, scalable, and fault tolerant.

If you want to learn basics about the Kafka, read the article Integrate Kafka Messaging Queue with NodeJS, in this article, you will find the basic of Kafka and its basic implementations.

In this article We will implement the Kafka Module in Nodejs using Typescript and NestJS.
We will create a service, this service will have the initialization and producer logic to produce the message in our given Kafka topic.

We will store our secrets in AWS Secrets Manager, we will connect our AWS secrets manager, and will fetch these secrets in our Kafka Module.

Later, will see how we can use this Kafka module in another module throughout our application.

Let's get started

Here we will create only two file in our module directory as.

Our module directory will look like this.
kafka/kafka.module.ts
kafka/producer.service.ts

And here we will use or import the Kafka Module and use it in our service to produce the messages.
app/app.module.ts
app/app.service.ts

Step 1: Create a Kafka Module

Create a file in our module directory ie kafka.module.ts.

File:kafka.module.ts

                    
                            import { Module } from '@nestjs/common';
                            import { KafkaProducerService } from 'src/kafka/producer.service';
                            
                            @Module({
                                providers: [KafkaProducerService],
                                exports: [KafkaProducerService]
                            })
                            export class KafkaModule { }
                            
                          

In the above code we have created a module in our kafka directory, this code creates a module and provide a producer service.

Step 2: Create a Kafka Producer

Create a file in our module directory ie kproducer.service.ts.

File:producer.service.ts

                    
                            import {
                                Injectable,
                                OnApplicationShutdown,
                                OnModuleInit,
                              } from '@nestjs/common';
                              import { Kafka, Producer, ProducerRecord } from 'kafkajs';
                              import { SecretsManager } from 'aws-sdk';
                              
                              @Injectable()
                              export class KafkaProducerService implements OnModuleInit, OnApplicationShutdown {
                                KafkaProducer: Producer;
                              
                                async onModuleInit() {
                                  try {
                                    let SecretsManagerClient = new SecretsManager({
                                      region: 'YOUR_SECRET_MANAGER_REGION',
                                    });
                                    const SecretsManagerResult = await SecretsManagerClient
                                      .getSecretValue({
                                        SecretId: 'YOUR_SECRET_MANAGER_KEY',
                                      })
                                      .promise();
                                    const SecretsManagerResponse = JSON.parse(SecretsManagerResult.SecretString);
                                    const kafka: Kafka = new Kafka({
                                      clientId: 'YOUR_KAFKA_CLIENT_ID',
                                      brokers: ["YOUR_KAFKA_BROKER_URL"],
                                      ssl: true,
                                      sasl: {
                                        mechanism: 'plain',
                                        username: SecretsManagerResponse.clsa_key,
                                        password: SecretsManagerResponse.clsa_secret,
                                      },
                                    });
                                    this.KafkaProducer = kafka.producer();
                                    await this.KafkaProducer.connect();
                                  } catch (error) {
                                    console.error(JSON.stringify(error));
                                  }
                                }
                              
                                async produce(record: ProducerRecord) {
                                  await this.KafkaProducer.send(record);
                                }
                              
                                async onApplicationShutdown() {
                                  await this.KafkaProducer.disconnect();
                                }
                              }                 
                    
                          

Step 3: Use this Kafka Module

File:app/app.module.ts

                            
                            import { Module } from "@nestjs/common/decorators";
                            import { KafkaModule } from "src/kafka/kafka.module";
                            import { KafkaProducerService } from "src/kafka/producer.service";
                            
                            @Module({
                                imports: [KafkaModule],
                                controllers: [],
                                providers: [KafkaProducerService],
                                exports: [KafkaProducerService],
                            })
                            export class AppModule {}
                            
                          

File:app/app.service.ts

                    
                            import { Injectable } from "@nestjs/common";
                            import { IHeaders, ProducerRecord } from "kafkajs";
                            import { ProducerService } from "src/kafka/producer.service";
                            
                            @Injectable()
                            export class AppService {
                              constructor(private readonly kafkaProducerService: ProducerService) { }
                            
                              async publishToKafka(id: string, data: any) {
                                const headers: IHeaders = {};
                                headers['some-header-value'] = 'some-header-value';
                                const msg = { id, data }
                                const record: ProducerRecord = {
                                  topic: 'YOUR_TOPIC_NAME',
                                  messages: [
                                    {
                                      value: JSON.stringify(msg),
                                      headers: headers,
                                    },
                                  ],
                                };
                                return await this.kafkaProducerService.produce(record);
                              }
                            }npm install express path
                            
                          

Now, only things you need to do is, to call publishToKafka('someId', {mydata:'mydata'})

publishToKafka('someId', { mydata:'mydata'})

---

Conclusion:

In this article, we have implemented Kafka Module with AWS Secrets Manager in NodeJS with TypeScript with the following things:
What is Apache Kafka
Create a Kafka Module
Create a Kafka Producer
Use the Kafka Module

Related Keywords:

Implement Kafka Module with AWS Secrets Manager in NodeJS with TypeScript

Kafka Client - AWS SDK for JavaScript

How do I use aws secret manager with nodejs

Integrate AWS Secret Manager and Kafka in your NodeJS application

Intro to KafkaJS - a Modern Kafka Client for Node.js